package com.zx.learn.flink.cdc;

//import com.ouyang.gmall.common.utils.FlinkUtil;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.api.Table;
//import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.types.Row;

/**https://blog.csdn.net/yang_shibiao/article/details/122774389
 * 基于SQL的CDC
 */
public class GmallCDCWithSQL {

//    public static void main(String[] args) throws Exception {
//
//        String applicationName = "gmall-cdc-sql";
//        long interval = 5000L;
//
//        // 1.获取执行环境，并配置checkpoint
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        FlinkUtil.deployRocksdbCheckpoint(env, applicationName, interval, true);
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//
//
//        //2.DDL方式建表
//        tableEnv.executeSql("CREATE TABLE mysql_binlog ( " +
//                " id STRING NOT NULL, " +
//                " tm_name STRING, " +
//                " logo_url STRING " +
//                ") WITH ( " +
//                " 'connector' = 'mysql-cdc', " +
//                " 'hostname' = 'bigdata1', " +
//                " 'port' = '3306', " +
//                " 'username' = 'root', " +
//                " 'password' = '123456', " +
//                " 'database-name' = 'gmall', " +
//                " 'table-name' = 'base_trademark' " +
//                ")");
//
//        //3.查询数据
//        Table table = tableEnv.sqlQuery("select * from mysql_binlog");
//
//        //4.将动态表转换为流
//        DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);
//        retractStream.print();
//
//        //5.启动任务
//        env.execute(applicationName);
//
//    }
}
